package cn.doitedu.api;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;
import java.util.HashMap;

public class _07_MyJdbcSourceFunction extends RichSourceFunction<String> {

    String jdbcUrl;
    String dbName;
    String tableName;
    String userName;
    String passWord;

    Connection connection;
    PreparedStatement stmt;
    ResultSet resultSet;


    int columnCount;
    HashMap<Integer, String> columnNames;

    // function的构造方法
    private _07_MyJdbcSourceFunction(String jdbcUrl, String dbName, String tableName, String userName, String passWord) throws SQLException {

        this.jdbcUrl = jdbcUrl;
        this.dbName = dbName;
        this.tableName = tableName;
        this.userName = userName;
        this.passWord = passWord;

    }


    // 这是 算子的工作生命周期方法：
    // open是在task调用算子做正式数据处理之前调用一次
    @Override
    public void open(Configuration parameters) throws Exception {

        this.connection = DriverManager.getConnection(jdbcUrl + dbName, userName, passWord);
        this.stmt = connection.prepareStatement("select * from " + tableName);
        this.resultSet = stmt.executeQuery();

        columnNames = new HashMap<>();

        // 获取查询结果的元数据
        ResultSetMetaData metaData = resultSet.getMetaData();
        // 从元数据中获取总共有几个列
        this.columnCount = metaData.getColumnCount();

        // 获取查询结果中的每一个列的列名
        for (int i = 1; i <= columnCount; i++) {
            String columnName = metaData.getColumnName(i);
            columnNames.put(i, columnName);
        }

    }


    // 这是 算子的工作生命周期方法：
    // run是在task调用source算子做数据读取的方法
    @Override
    public void run(SourceContext<String> ctx) throws Exception {

        JSONObject record = new JSONObject();

        // 一次遍历，指向一行结果
        while (resultSet.next()) {

            // 在一行结果中，循环获取每个字段
            for (int i = 1; i <= columnCount; i++) {
                // 取第i字段的字段名
                String columnName = columnNames.get(i);

                // 取第i字段的字段值
                Object columnValue = resultSet.getObject(i);

                // 把取出的字段放入json对象
                record.put(columnName, columnValue);
            }

            // 把本行的结果变成json返回
            ctx.collect(record.toJSONString());

        }
    }


    // 这是 算子的工作生命周期方法：
    // task结束时，会调用算子的close方法
    @Override
    public void close() throws Exception {

        resultSet.close();
        stmt.close();
        connection.close();

    }

    @Override
    public void cancel() {


    }

    public static Builder builder() {
        return new Builder();
    }

    public static class Builder {

        String jdbcUrl;
        String userName;
        String password;
        String dbName;
        String tableName;

        public Builder setJdbcUrl(String jdbcUrl) {
            this.jdbcUrl = jdbcUrl;
            return this;
        }


        public Builder setDbUserName(String userName) {
            this.userName = userName;
            return this;
        }

        public Builder setDbPassword(String password) {
            this.password = password;
            return this;
        }


        public Builder setDbName(String dbName) {
            this.dbName = dbName;
            return this;
        }

        public Builder setTableName(String tableName) {
            this.tableName = tableName;
            return this;
        }


        public _07_MyJdbcSourceFunction build() throws SQLException {

            return new _07_MyJdbcSourceFunction(jdbcUrl, dbName, tableName, userName, password);
        }

    }


}
